package com.shujia.search

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object ScopeSearchCar {
  def main(args: Array[String]): Unit = {
    val configuration = new Configuration()
    configuration.setLong("akka.ask.timeout", 100000)

    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment(1, configuration)


    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //使用流模型
      .build()

    //创建flink sql的执行环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)


    /**
      * \、读取过车数据
      *
      */


    bsTableEnv.executeSql(
      """
        |CREATE TABLE cars_kafka (
        | car String,
        | city_code String,
        | county_code String,
        | card BIGINT,
        | camera_id String,
        | orientation String,
        | road_id BIGINT,
        | `time` BIGINT,
        | speed Double
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'cars',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'json',
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
      """.stripMargin)


    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE cards_hbase (
        | card_id STRING,
        | info ROW<lat DOUBLE,lon DOUBLE>,
        | PRIMARY KEY (card_id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'cards',
        | 'zookeeper.quorum' = 'master:2181,node1:2181,node2:2181'
        |)
        |
      """.stripMargin)


    /*

PUT /cars/
{
  "mappings": {
    "properties": {
      "ts": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss"
      },
      "card_location": {
        "type": "geo_point"
      }
    }
  }
}

     */


    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE cars_es (
        | car String,
        | city_code String,
        | county_code String,
        | card BIGINT,
        | camera_id String,
        | orientation String,
        | road_id BIGINT,
        | `time` BIGINT,
        | speed Double,
        | card_location array<Double>,
        | ts  String,
        |  PRIMARY KEY (car,`time`) NOT ENFORCED
        |) WITH (
        |  'connector' = 'elasticsearch-7',
        |  'hosts' = 'http://master:9200',
        |  'index' = 'cars'
        |)
        |
      """.stripMargin)



    bsTableEnv.executeSql(
      """
        |
        |insert into cars_es
        |select car,city_code,county_code,card,camera_id,orientation,road_id,`time`,speed,
        |array[ b.info.lon,b.info.lat] as card_location,
        |FROM_UNIXTIME(`time`) as ts
        |from
        |cars_kafka  as a join cards_hbase  as b
        |on
        |a.card = cast(b.card_id as BIGINT)
        |
      """.stripMargin)


  }

}
